package com.test.finkcdc.util;

import com.alibaba.fastjson.JSONObject;
import com.test.finkcdc.config.TargetConfig;
import com.test.finkcdc.entity.JobInfo;
import oracle.sql.TIMESTAMP;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.Timestamp;

public class UpdateSink extends RichSinkFunction implements Serializable {
    private static final long serialVersionUID = 1256200277148844927L;
    private Connection conn;
    private PreparedStatement pre;
    private final JobInfo info;

    public UpdateSink(JobInfo info) {
        this.info = info;
    }

    public void open(Configuration parameters) throws Exception {
        conn = TargetConfig.dataSource.getConnection();
        pre = conn.prepareStatement(info.getDeleteSql());
    }

    public void invoke(Object paramValue, SinkFunction.Context context) throws Exception {
        JSONObject jsonObject = JSONObject.parseObject(paramValue + "");
        Object obj = JSONObject.parseObject(jsonObject.getString("jsonBean"), info.getEntity());
        Field[] fields = obj.getClass().getDeclaredFields();
        for (int i = 2; i < fields.length; i++) {
            Field field = fields[i];
            field.setAccessible(true);
            DateFormatMark dateFormatMark = field.getAnnotation(DateFormatMark.class);
            Object value = field.get(obj);
            if (dateFormatMark != null && value != null) {
                pre.setObject(i - 1, new Timestamp((Long) value - 8 * 60 * 60 * 1000));
            } else {
                pre.setObject(i - 1, value);
            }
        }
        Field field = obj.getClass().getDeclaredField(info.getTableId());
        field.setAccessible(true);
        pre.setObject(fields.length - 1, field.get(obj));
        pre.executeUpdate();
        conn.setAutoCommit(false);
        conn.commit();
    }

    public void close() throws Exception {
        super.close();
        if (pre != null) {
            pre.close();
        }
        if (conn != null) {
            conn.close();
        }
    }
}
